dbtのベストプラクティスをより理解するために、dbt-athenaを使ってGeneric Testを作ってみた
こんちには。
データアナリティクス事業本部 インテグレーション部 機械学習チームの中村です。
前回に引き続き、dbtのベストプラクティスをより理解するために、dbt-athenaを使って新規プロジェクトからdbtのベストプラクティスに寄せてみたいと思います。
今回はtestについて調べた上で、Public Templateに沿ったテストを組み込んでいきたいと思います。
ここまでは以下の記事を参照ください。
テストについて
テスト概要
テストに関する記載は以下にあります。
この記載によれば、テストは大きく分けて２種類に分かれています。
- Singular Test
- 単発のテスト
tests/{test_name}.sqlで記載され、件数が0件であれば成功とするようにクエリを記載する
- Generic Test
- 汎用的に繰り返し使用可能なテスト
- dbtには、
unique、not_null、accepted_values、relationshipの4つのGeneric Testがすでに定義済み（これらがスキーマテストと呼ばれていた） - カスタムとして、テスト用のマクロをテストブロックの中で定義し、それらをymlファイルのModel Propertiesで呼び出すことでテストを実行も可能
このうち、本稿ではGeneric Testの方を取り扱います。
Generic Testの記述方法
Generic Testを記述するには、Model Propertiesにtestsブロックを記載します。
（他にも方法はありますが、一番よく見る形）
testsブロックの記法については以下に詳細が記載してあります。
以下がそのフォーマットです。
version: 2
models:
- name: <model_name>
tests:
- <test_name>:
<argument_name>: <argument_value>
config:
<test_config>: <config-value>
columns:
- name: <column_name>
tests:
- <test_name>
- <test_name>:
<argument_name>: <argument_value>
config:
<test_config>: <config-value>
モデルレベルのテスト記述と、カラムレベルのテスト記述が可能です。
argument_nameはテスト名ですが、未指定の場合は自動で生成されます。
configはwhereの例がよく登場しますが、設定可能な値は以下に記載されています。
テストの実行方法
コマンドとしてはdbt testでテストを実行することが可能ですが、様々な指定方法でテスト対象をselectすることが可能となっています。
以下に概要が書いてあります。
より細かいselectについては以下にも記載がありました。
test_type毎（singular、generic）の指定、サブフォルダの指定、モデルの指定、ソーステーブルの指定、タグの指定など様々なselectが準備されているようです。
カスタムGeneric Testの定義方法
カスタムGeneric Testの定義方法は以下に記載があります。
tests/generic/配下にsqlファイルを記載することで、カスタムGeneric Testが定義ができます。
（公式ドキュメントでは例として偶数かどうかのテストが挙げられていました）
{% test is_even(model, column_name) %}
-- ここにSQLでテストを書く
{% endtest %}
こちらをModel Propertiesで呼び出すこと（以下ではis_evenが該当）でテストを実行することが可能となります。
version: 2
models:
- name: users
columns:
- name: favorite_number
tests:
- is_even
やってみた
これらの知識を前提にテストを作っていきます。
前回までの記事の以下は終わっている前提とします。
- dbt実行環境の準備
- dbtプロジェクト作成
- dbt seedでソーステーブルを作成
- stagingレイヤのモデルの追加
- martsレイヤのモデル作成
stagingレイヤにテストを追加
Public Templateの以下にstagingレイヤのモデルのymlファイルが格納されています。
（tests以外のdescriptionなども含まれますが）
こちらから以下のファイルをmodels/stagingフォルダに配置します。
stg_customers.sql stg_locations.sql stg_order_items.sql stg_orders.sql stg_products.sql stg_supplies.sql
テスト内容は定義済みのGeneric Testとしてnot_nullやuniqueなどのテストを実行するものとなっています。（以下一例）
models:
- name: stg_customers
description: Customer data with basic cleaning and transformation applied, one row per customer.
columns:
- name: customer_id
description: The unique key for each customer.
tests:
- not_null
- unique
テストを以下で実行します。
dbt test --select staging # 10:30:57 Running with dbt=1.6.7 # 10:30:58 Registered adapter: athena=1.6.4 # 10:30:58 Found 12 models, 6 seeds, 12 tests, 6 sources, 0 exposures, 0 metrics, 624 macros, 0 groups, 0 semantic models # 10:30:58 # 10:31:03 Concurrency: 1 threads (target='dev') # 10:31:03 # 10:31:03 1 of 12 START test not_null_stg_customers_customer_id .......................... [RUN] # 10:31:05 1 of 12 PASS not_null_stg_customers_customer_id ................................ [PASS in 2.01s] # 10:31:05 2 of 12 START test not_null_stg_locations_location_id .......................... [RUN] # 10:31:07 2 of 12 PASS not_null_stg_locations_location_id ................................ [PASS in 2.01s] # 10:31:07 3 of 12 START test not_null_stg_order_items_order_item_id ...................... [RUN] # 10:31:09 3 of 12 PASS not_null_stg_order_items_order_item_id ............................ [PASS in 2.24s] # 10:31:09 4 of 12 START test not_null_stg_orders_order_id ................................ [RUN] # 10:31:12 4 of 12 PASS not_null_stg_orders_order_id ...................................... [PASS in 3.02s] # 10:31:12 5 of 12 START test not_null_stg_products_product_id ............................ [RUN] # 10:31:14 5 of 12 PASS not_null_stg_products_product_id .................................. [PASS in 2.05s] # 10:31:14 6 of 12 START test not_null_stg_supplies_supply_uuid ........................... [RUN] # 10:31:16 6 of 12 PASS not_null_stg_supplies_supply_uuid ................................. [PASS in 1.96s] # 10:31:16 7 of 12 START test unique_stg_customers_customer_id ............................ [RUN] # 10:31:18 7 of 12 PASS unique_stg_customers_customer_id .................................. [PASS in 1.92s] # 10:31:18 8 of 12 START test unique_stg_locations_location_id ............................ [RUN] # 10:31:20 8 of 12 PASS unique_stg_locations_location_id .................................. [PASS in 2.11s] # 10:31:20 9 of 12 START test unique_stg_order_items_order_item_id ........................ [RUN] # 10:31:23 9 of 12 PASS unique_stg_order_items_order_item_id .............................. [PASS in 2.91s] # 10:31:23 10 of 12 START test unique_stg_orders_order_id ................................. [RUN] # 10:31:26 10 of 12 PASS unique_stg_orders_order_id ....................................... [PASS in 2.98s] # 10:31:26 11 of 12 START test unique_stg_products_product_id ............................. [RUN] # 10:31:28 11 of 12 PASS unique_stg_products_product_id ................................... [PASS in 2.07s] # 10:31:28 12 of 12 START test unique_stg_supplies_supply_uuid ............................ [RUN] # 10:31:30 12 of 12 PASS unique_stg_supplies_supply_uuid .................................. [PASS in 1.88s] # 10:31:30 # 10:31:30 Finished running 12 tests in 0 hours 0 minutes and 32.29 seconds (32.29s). # 10:31:30 # 10:31:30 Completed successfully # 10:31:30 # 10:31:30 Done. PASS=12 WARN=0 ERROR=0 SKIP=0 TOTAL=12
statingレイヤのモデルについてテストが実行できました。
martsレイヤにテストを追加
Public Templateの以下にmartsレイヤのモデルのymlファイルが格納されています。
（こちらもtests以外のdescriptionなども含まれますが）
こちらから以下のファイルをmodels/martsフォルダに配置します。
customers.yml locations.yml order_items.yml orders.yml products.yml supplies.yml
これらのymlファイルから、semantic_modelsブロックとmetricsブロックは今回使用しないため削除します。
テスト内容は定義済みのGeneric Testとしてnot_nullやunique、accepted_valuesなどのテストを実行するものとなっています。（以下一例）
models:
- name: customers
description: Customer overview data mart, offering key details for each unique customer. One row per customer.
columns:
- name: customer_id
description: The unique key of the orders mart.
tests:
- not_null
- unique
- name: customer_name
description: Customers' full name.
- name: count_lifetime_orders
description: Total number of orders a customer has ever placed.
- name: first_ordered_at
description: The timestamp when a customer placed their first order.
- name: last_ordered_at
description: The timestamp of a customer's most recent order.
- name: lifetime_spend_pretax
description: The sum of all the pre-tax subtotals of every order a customer has placed.
- name: lifetime_spend
description: The sum of all the order totals (including tax) that a customer has ever placed.
- name: customer_type
description: Options are 'new' or 'returning', indicating if a customer has ordered more than once or has only placed their first order to date.
tests:
- accepted_values:
values: ["new", "returning"]
テストを以下で実行します。
dbt test --select marts # 10:41:59 Running with dbt=1.6.7 # 10:41:59 Registered adapter: athena=1.6.4 # 10:42:00 Found 12 models, 6 seeds, 20 tests, 6 sources, 0 exposures, 0 metrics, 624 macros, 0 groups, 0 semantic models # 10:42:00 # 10:42:04 Concurrency: 1 threads (target='dev') # 10:42:04 # 10:42:04 1 of 8 START test accepted_values_customers_customer_type__new__returning ...... [RUN] # 10:42:07 1 of 8 PASS accepted_values_customers_customer_type__new__returning ............ [PASS in 3.00s] # 10:42:07 2 of 8 START test not_null_customers_customer_id ............................... [RUN] # 10:42:10 2 of 8 PASS not_null_customers_customer_id ..................................... [PASS in 3.02s] # 10:42:10 3 of 8 START test not_null_order_items_order_item_id ........................... [RUN] # 10:42:12 3 of 8 PASS not_null_order_items_order_item_id ................................. [PASS in 1.93s] # 10:42:12 4 of 8 START test not_null_orders_order_id ..................................... [RUN] # 10:42:14 4 of 8 PASS not_null_orders_order_id ........................................... [PASS in 1.97s] # 10:42:14 5 of 8 START test relationships_orders_customer_id__customer_id__ref_stg_customers_ [RUN] # 10:42:16 5 of 8 PASS relationships_orders_customer_id__customer_id__ref_stg_customers_ .. [PASS in 2.03s] # 10:42:16 6 of 8 START test unique_customers_customer_id ................................. [RUN] # 10:42:18 6 of 8 PASS unique_customers_customer_id ....................................... [PASS in 2.01s] # 10:42:18 7 of 8 START test unique_order_items_order_item_id ............................. [RUN] # 10:42:20 7 of 8 PASS unique_order_items_order_item_id ................................... [PASS in 1.97s] # 10:42:20 8 of 8 START test unique_orders_order_id ....................................... [RUN] # 10:42:23 8 of 8 PASS unique_orders_order_id ............................................. [PASS in 3.03s] # 10:42:23 # 10:42:23 Finished running 8 tests in 0 hours 0 minutes and 23.81 seconds (23.81s). # 10:42:23 # 10:42:23 Completed successfully # 10:42:23 # 10:42:23 Done. PASS=8 WARN=0 ERROR=0 SKIP=0 TOTAL=8
martsレイヤのモデルについてテストが実行できました。
relationshipのテストを実装する
定義済みのGeneric Testのうちrelationshipが、Public Templateでは実装されていなかったので、追加してみます。
relationshipは参照整合性のテストとなり、以下のようなものとなります。（stg_ordersモデルの例）
models:
- name: stg_orders
description: Order data with basic cleaning and transformation applied, one row per order.
columns:
- name: order_id
description: The unique key for each order.
tests:
- not_null
- unique
- name: location_id
tests:
- relationships:
to: ref('stg_locations')
field: location_id
config:
tags: 'test_relationship'
- name: customer_id
tests:
- relationships:
to: ref('stg_customers')
field: customer_id
config:
tags: 'test_relationship'
要するにstg_ordersモデルのcustomer_idは、きちんとstg_customersモデルのcustomer_idに含まれるかどうかをテストできます。
また今回、relationshipのテストを単体で実行できるようにtags: 'test_relationship'も追加してみました。
こちらを同様にrelationshipのテストが必要そうなstg_order_itemsモデルとstg_suppliesモデルにも追加します。
そしてテストを以下で実行します。
dbt test --select "tag:test_relationship" # 11:23:23 Running with dbt=1.6.7 # 11:23:23 Registered adapter: athena=1.6.4 # 11:23:23 Found 12 models, 6 seeds, 25 tests, 6 sources, 0 exposures, 0 metrics, 624 macros, 0 groups, 0 semantic models # 11:23:23 # 11:23:28 Concurrency: 1 threads (target='dev') # 11:23:28 # 11:23:28 1 of 5 START test relationships_stg_order_items_order_id__order_id__ref_stg_orders_ [RUN] # 11:23:31 1 of 5 PASS relationships_stg_order_items_order_id__order_id__ref_stg_orders_ .. [PASS in 3.07s] # 11:23:31 2 of 5 START test relationships_stg_order_items_product_id__product_id__ref_stg_products_ [RUN] # 11:23:34 2 of 5 PASS relationships_stg_order_items_product_id__product_id__ref_stg_products_ [PASS in 3.00s] # 11:23:34 3 of 5 START test relationships_stg_orders_customer_id__customer_id__ref_stg_customers_ [RUN] # 11:23:37 3 of 5 PASS relationships_stg_orders_customer_id__customer_id__ref_stg_customers_ [PASS in 2.90s] # 11:23:37 4 of 5 START test relationships_stg_orders_location_id__location_id__ref_stg_locations_ [RUN] # 11:23:40 4 of 5 PASS relationships_stg_orders_location_id__location_id__ref_stg_locations_ [PASS in 3.20s] # 11:23:40 5 of 5 START test relationships_stg_supplies_product_id__product_id__ref_stg_products_ [RUN] # 11:23:42 5 of 5 PASS relationships_stg_supplies_product_id__product_id__ref_stg_products_ [PASS in 1.95s] # 11:23:42 # 11:23:42 Finished running 5 tests in 0 hours 0 minutes and 18.67 seconds (18.67s). # 11:23:42 # 11:23:42 Completed successfully # 11:23:42
無事にrelationshipのテストのみ動かすことができました。
変更点の詳細は以下のcommit参照してください。（他にもテストを追加できる部分は残っていると思います）
カスタムGeneric Testの追加
正の値しか取らないと考えられるカラムも多くありそうでしたので、これを題材にカスタムGeneric Testの追加もやってみます。
tests/generic/test_is_plus.sqlというファイルを作成して、以下を記載します。
{% test is_plus(model, column_name) %}
with validation as (
select
{{ column_name }} as target_field
from {{ model }}
),
validation_errors as (
select
target_field
from validation
-- if this is true, then target_field is plus value!
where target_field < 0
)
select *
from validation_errors
{% endtest %}
再度、stg_ordersモデルの例として以下のようにis_plusをGeneric Testとして追加します。（tags: 'test_is_plus'も付与）
models:
- name: stg_orders
description: Order data with basic cleaning and transformation applied, one row per order.
columns:
- name: order_id
description: The unique key for each order.
tests:
- not_null
- unique
- name: location_id
tests:
- relationships:
to: ref('stg_locations')
field: location_id
config:
tags: 'test_relationship'
- name: customer_id
tests:
- relationships:
to: ref('stg_customers')
field: customer_id
config:
tags: 'test_relationship'
- name: order_total
tests:
- is_plus:
config:
tags: 'test_is_plus'
- name: tax_paid
tests:
- is_plus:
config:
tags: 'test_is_plus'
そしてテストを以下で実行します。
dbt test --select "tag:test_is_plus" # 11:41:37 Running with dbt=1.6.7 # 11:41:38 Registered adapter: athena=1.6.4 # 11:41:38 Found 12 models, 6 seeds, 27 tests, 6 sources, 0 exposures, 0 metrics, 625 macros, 0 groups, 0 semantic models # 11:41:38 # 11:41:43 Concurrency: 1 threads (target='dev') # 11:41:43 # 11:41:43 1 of 2 START test is_plus_stg_orders_order_total ............................... [RUN] # 11:41:45 1 of 2 PASS is_plus_stg_orders_order_total ..................................... [PASS in 2.14s] # 11:41:45 2 of 2 START test is_plus_stg_orders_tax_paid .................................. [RUN] # 11:41:47 2 of 2 PASS is_plus_stg_orders_tax_paid ........................................ [PASS in 1.99s] # 11:41:47 # 11:41:47 Finished running 2 tests in 0 hours 0 minutes and 8.97 seconds (8.97s). # 11:41:47 # 11:41:47 Completed successfully # 11:41:47 # 11:41:47 Done. PASS=2 WARN=0 ERROR=0 SKIP=0 TOTAL=2
無事に正の値をとるかどうかのテストを動かすことができました。
変更点の詳細は以下のcommit参照してください。（他にもテストを追加できる部分は残っていると思います）
ソーステーブルへテストを実装する
テストはmodelsだけではなく、ソーステーブルに対しても実装することが可能です。
以下のように__sources.ymlにGeneric Testを追加してみます。
version: 2
sources:
- name: ecom
schema: jaffle_shop_raw
description: E-commerce data for the Jaffle Shop
tables:
- name: raw_customers
description: One record per person who has purchased one or more items
columns:
- name: id
tests:
- not_null
- unique
- name: raw_orders
description: One record per order (consisting of one or more order items)
columns:
- name: id
tests:
- not_null
- unique
- name: raw_items
description: Items included in an order
columns:
- name: id
tests:
- not_null
- unique
- name: raw_stores
description: Items included in an order
columns:
- name: id
tests:
- not_null
- unique
- name: raw_products
description: One record per SKU for items sold in stores
columns:
- name: sku
tests:
- not_null
- unique
- name: raw_supplies
description: One record per supply per SKU of items sold in stores
そしてテストを以下で実行します。（ソーステーブルのみの指定がselectで可能です）
dbt test --select "source:*" # 11:58:04 Running with dbt=1.6.7 # 11:58:04 Registered adapter: athena=1.6.4 # 11:58:04 Found 6 seeds, 12 models, 37 tests, 6 sources, 0 exposures, 0 metrics, 625 macros, 0 groups, 0 semantic models # 11:58:04 # 11:58:09 Concurrency: 1 threads (target='dev') # 11:58:09 # 11:58:09 1 of 10 START test source_not_null_ecom_raw_customers_id ....................... [RUN] # 11:58:11 1 of 10 PASS source_not_null_ecom_raw_customers_id ............................. [PASS in 2.13s] # 11:58:11 2 of 10 START test source_not_null_ecom_raw_items_id ........................... [RUN] # 11:58:13 2 of 10 PASS source_not_null_ecom_raw_items_id ................................. [PASS in 2.12s] # 11:58:13 3 of 10 START test source_not_null_ecom_raw_orders_id .......................... [RUN] # 11:58:16 3 of 10 PASS source_not_null_ecom_raw_orders_id ................................ [PASS in 2.98s] # 11:58:16 4 of 10 START test source_not_null_ecom_raw_products_sku ....................... [RUN] # 11:58:19 4 of 10 PASS source_not_null_ecom_raw_products_sku ............................. [PASS in 3.03s] # 11:58:19 5 of 10 START test source_not_null_ecom_raw_stores_id .......................... [RUN] # 11:58:21 5 of 10 PASS source_not_null_ecom_raw_stores_id ................................ [PASS in 1.98s] # 11:58:21 6 of 10 START test source_unique_ecom_raw_customers_id ......................... [RUN] # 11:58:23 6 of 10 PASS source_unique_ecom_raw_customers_id ............................... [PASS in 1.97s] # 11:58:23 7 of 10 START test source_unique_ecom_raw_items_id ............................. [RUN] # 11:58:25 7 of 10 PASS source_unique_ecom_raw_items_id ................................... [PASS in 1.93s] # 11:58:25 8 of 10 START test source_unique_ecom_raw_orders_id ............................ [RUN] # 11:58:27 8 of 10 PASS source_unique_ecom_raw_orders_id .................................. [PASS in 2.09s] # 11:58:27 9 of 10 START test source_unique_ecom_raw_products_sku ......................... [RUN] # 11:58:30 9 of 10 PASS source_unique_ecom_raw_products_sku ............................... [PASS in 2.95s] # 11:58:30 10 of 10 START test source_unique_ecom_raw_stores_id ........................... [RUN] # 11:58:32 10 of 10 PASS source_unique_ecom_raw_stores_id ................................. [PASS in 1.98s] # 11:58:32 # 11:58:32 Finished running 10 tests in 0 hours 0 minutes and 27.84 seconds (27.84s). # 11:58:32 # 11:58:32 Completed successfully # 11:58:32 # 11:58:32 Done. PASS=10 WARN=0 ERROR=0 SKIP=0 TOTAL=10
無事にソーステーブルのテストを動かすことができました。
変更点の詳細は以下のcommit参照してください。
補足 : モデル作成前にGeneric Testは実行できない
テスト対象のモデルを作成する前にGeneric Testを実行することはできないようですので注意が必要です。
試しにstg_customersをいったん削除して、以下のようにテストをselectで実行してみます。
dbt test --select "stg_customers" # 12:02:32 Running with dbt=1.6.7 # 12:02:32 Registered adapter: athena=1.6.4 # 12:02:32 Found 6 seeds, 12 models, 37 tests, 6 sources, 0 exposures, 0 metrics, 625 macros, 0 groups, 0 semantic models # 12:02:32 # 12:02:37 Concurrency: 1 threads (target='dev') # 12:02:37 # 12:02:37 1 of 4 START test not_null_stg_customers_customer_id ........................... [RUN] # 12:05:52 1 of 4 ERROR not_null_stg_customers_customer_id ................................ [ERROR in 195.00s] # ...(以下略)...
エラーとなるまでもかなり時間がかかりました。Athenaの画面を確認すると、6回リトライを試みているようです。
あまりテストが進まないなと思ったら、足りないテーブルがないかどうかもマネジメントコンソールで確認した方がよさそうですね。
ここまでのレポジトリ
ここまでを実行したGitHubレポジトリを以下に公開しています。
まとめ
いかがでしたでしょうか。本記事がdbt-athenaでdbtを始められる方の参考になれば幸いです。
